#include "my_cache_processor.h"
#include "my_cache_layer.h"
#include "comm/my_syscall.h"
#include "disk/my_disk_cache.h"
#include "event/my_event_processor.h"

static const int my_mem_cache_level     = 0;
static const int my_fast_cache_level    = 1;
static const int my_storage_cache_level = 2;

typedef enum {
    CACHE_PROCESSOR_UNINIT          = 0,
    CACHE_PROCESSOR_STARTING        = 1,
    CACHE_PROCESSOR_STATED          = 2,
    CACHE_PROCESSOR_CLOSE_WRITE     = 3,
    CACHE_PROCESSOR_CLOSE           = 4,
} my_cache_processor_stats_t;

typedef struct {
    my_disk_cache_t                 disk_cache;
    my_bool_t                       open;
    my_dispatch_event_t             dispatch_event;
    int                             level;
    my_cache_config_disk_t          *disk_cfg;
    my_bool_t                       create_new;
} my_disk_cache_node_t;

static inline void
my_disk_cache_node_init(my_disk_cache_node_t *node, int level,
    my_cache_config_disk_t *disk_cfg, my_bool_t create_new) {
    bzero(node, sizeof(my_disk_cache_node_t));
    my_disk_cache_init(&node->disk_cache);
    // no need to do this
    // my_dispatch_event_init(&node->dispatch_event);
    node->level = level;
    node->disk_cfg = disk_cfg;
    node->create_new = create_new;
}

typedef struct my_cache_processor_s my_cache_processor_t;

struct my_cache_processor_s {
    my_cache_processor_stats_t      stats;
    my_bool_t                       close_write;
    my_cache_mem_layer_t            *mem_cache;
    my_cache_storage_layer_t        *fast_cache;    // L1
    my_cache_storage_layer_t        *storage_cache; // L2

    // for disk cache initialization
    my_disk_cache_node_t            *fast_disk_cache_nodes;
    my_disk_cache_node_t            *storage_disk_cache_nodes;

    // for async initialization
    my_mutex_t                      mutex;
    my_cond_t                       cond;
    int                             fast_disk_open_num;
    int                             storage_disk_open_num;
};

static my_mutex_t guard = MY_MUTEX_INITIALIZER;
static my_cache_processor_t g_cache_processor = {
    .stats                      = CACHE_PROCESSOR_UNINIT,
    .mem_cache                  = NULL,
    .fast_cache                 = NULL,
    .storage_cache              = NULL,
    .fast_disk_cache_nodes      = NULL,
    .storage_disk_cache_nodes   = NULL,
    .mutex                      = MY_MUTEX_INITIALIZER,
    .cond                       = MY_COND_INITIALIZER,
    .fast_disk_open_num         = 0,
    .storage_disk_open_num      = 0
};

static void
my_cache_processor_kicked_handler(my_disk_volume_t *volume) {
    my_cache_storage_layer_t *layer = g_cache_processor.fast_cache;
    if (volume->level != my_fast_cache_level) {
        layer = g_cache_processor.storage_cache;
    }

    my_cache_storage_layer_kicked(layer, volume);
}

static void
my_cache_processor_check_start_done(int level) {
    my_bool_t all_done = MY_FALSE;

    int n_fast_disk = g_cache_conf.storage_cache.n_fast_disk;
    int n_storage_disk = g_cache_conf.storage_cache.n_storage_disk;

    my_mutex_lock(&g_cache_processor.mutex);

    if (level = my_fast_cache_level) {
        g_cache_processor.fast_disk_open_num++;
    } else {
        g_cache_processor.storage_disk_open_num++;
    }

    if (g_cache_processor.fast_disk_open_num == n_fast_disk &&
        g_cache_processor.storage_disk_open_num == n_storage_disk) {
        my_cache_storage_layer_build(g_cache_processor.fast_cache);
        my_cache_storage_layer_build(g_cache_processor.storage_cache);
        my_cond_signal(&g_cache_processor.cond);
    }

    my_mutex_unlock(&g_cache_processor.mutex);
}

static void
my_cache_processor_fast_layer_open_disk_done(my_disk_cache_t *disk_cache, my_bool_t success) {
    if (success) {
        for (int i = 0; i < disk_cache->nvolumes; i++) {
            my_cache_storage_layer_add_storage(g_cache_processor.fast_cache,
                disk_cache->volumes + i, *(disk_cache->open_stats + i));
        }
    }

    my_cache_processor_check_start_done(my_fast_cache_level);
}

static void
my_cache_processor_storage_layer_open_disk_done(my_disk_cache_t *disk_cache, my_bool_t success) {
    if (success) {
        for (int i = 0; i < disk_cache->nvolumes; i++) {
            my_cache_storage_layer_add_storage(g_cache_processor.storage_cache,
                disk_cache->volumes + i, *(disk_cache->open_stats + i));
        }
    }

    my_cache_processor_check_start_done(my_storage_cache_level);
}

static void
my_cache_processor_open_disk_done(my_disk_cache_t *disk_cache, my_bool_t success) {
    my_disk_cache_node_t *node = disk_cache->ctx;
    int level = node->level;
    node->open = success;

    if (level == my_fast_cache_level) {
        my_cache_processor_fast_layer_open_disk_done(disk_cache, success);
        return;
    }

    my_cache_processor_storage_layer_open_disk_done(disk_cache, success);
}

static void
my_cache_processor_open_disk_handler(my_dispatch_event_t *event) {
    my_disk_cache_node_t *node = (my_disk_cache_node_t *)event->data;

    my_cache_config_disk_t *disk_cfg = node->disk_cfg;
    my_disk_cache_open(&node->disk_cache, node->level,
        disk_cfg->file_path, disk_cfg->except_size, node->create_new,
        my_cache_processor_kicked_handler,
        my_cache_processor_open_disk_done, node);
}

static void
my_cache_processor_open_fast_cache_layer(my_bool_t create_new) {
    g_cache_processor.fast_cache = my_malloc(sizeof(my_cache_storage_layer_t));
    my_cache_storage_layer_init(g_cache_processor.fast_cache);

    int ndisk = g_cache_conf.storage_cache.n_fast_disk;
    my_cache_config_disk_t *disks_cfg = g_cache_conf.storage_cache.fast_disks;

    g_cache_processor.fast_disk_cache_nodes = my_malloc(sizeof(my_disk_cache_node_t) * ndisk);
    for (int i = 0; i < ndisk; i++) {
        my_disk_cache_node_t *node = g_cache_processor.fast_disk_cache_nodes + i;
        my_disk_cache_node_init(node, 0, disks_cfg+i, create_new);
        // dispatch event
        node->dispatch_event.data = node;
        node->dispatch_event.handler = my_cache_processor_open_disk_handler;
        my_event_processor_post_event(&node->dispatch_event);
    }
}

static void
my_cache_processor_open_storage_cache_layer(my_bool_t create_new) {
    g_cache_processor.storage_cache = my_malloc(sizeof(my_cache_storage_layer_t));
    my_cache_storage_layer_init(g_cache_processor.storage_cache);

    int ndisk = g_cache_conf.storage_cache.n_storage_disk;
    my_cache_config_disk_t *disks_cfg = g_cache_conf.storage_cache.storage_disks;

    g_cache_processor.storage_disk_cache_nodes = my_malloc(sizeof(my_disk_cache_node_t) * ndisk);
    for (int i = 0; i < ndisk; i++) {
        my_disk_cache_node_t *node = g_cache_processor.storage_disk_cache_nodes + i;
        my_disk_cache_node_init(node, 1, disks_cfg+i, create_new);
        // dispatch event
        node->dispatch_event.data = node;
        node->dispatch_event.handler = my_cache_processor_open_disk_handler;
        my_event_processor_post_event(&node->dispatch_event);
    }
}

void
my_cache_processor_check_and_report() {
    int n_fast_disk = g_cache_conf.storage_cache.n_fast_disk;
    int n_storage_disk = g_cache_conf.storage_cache.n_storage_disk;

    my_disk_cache_node_t *node;

    int online = 0;

    for (int i = 0; i < n_fast_disk; i++) {
        node = g_cache_processor.fast_disk_cache_nodes + i;
        if (node->open) {
            online++;
            MY_LOG_DEBUG("cache_processor", "open %s cache success", node->disk_cfg->file_path);
        } else {
            MY_LOG_ERROR("cache_processor", "open %s cache failed", node->disk_cfg->file_path);
        }
    }

    if (online == 0) {
        MY_LOG_FATAL0("cache_processor", "fast cache is all offline, plrease check and fix");
    }

    online = 0;

    for (int i = 0; i < n_storage_disk; i++) {
        node = g_cache_processor.storage_disk_cache_nodes + i;
        if (node->open) {
            online++;
            MY_LOG_DEBUG("cache_processor", "open %s cache success", node->disk_cfg->file_path);
        } else {
            MY_LOG_ERROR("cache_processor", "open %s cache failed", node->disk_cfg->file_path);
        }
    }

    if (online == 0) {
        MY_LOG_FATAL0("cache_processor", "storage cache is all offline, plrease check and fix");
    }
}

void
my_cache_processor_start(my_bool_t create_new) {
    my_mutex_lock(&guard);
    if (g_cache_processor.stats != CACHE_PROCESSOR_UNINIT) {
        MY_LOG_FATAL0("cache_processor", "unexcept cache processor stats");
    }
    g_cache_processor.stats = CACHE_PROCESSOR_STARTING;
    my_mutex_unlock(&guard);

    my_cache_config_mem_layer_t *mem_cache_cfg = &g_cache_conf.mem_cache;
    if (mem_cache_cfg->enabled) {
        g_cache_processor.mem_cache = my_malloc(sizeof(my_cache_mem_layer_t));
        my_cache_mem_layer_open(g_cache_processor.mem_cache, mem_cache_cfg->nlist,
            mem_cache_cfg->max_item, mem_cache_cfg->max_bytes);
    }

    my_cache_config_storage_layer_t *storage_cache_cfg = &g_cache_conf.storage_cache;
    if (storage_cache_cfg->n_fast_disk > 0) {
        my_cache_processor_open_fast_cache_layer(create_new);
    }

    if (storage_cache_cfg->n_storage_disk == 0) {
        MY_LOG_FATAL0("cache_processor", "the number of storage disk must be greater than 0");
    }

    my_cache_processor_open_storage_cache_layer(create_new);

    int n_fast_disk = g_cache_conf.storage_cache.n_fast_disk;
    int n_storage_disk = g_cache_conf.storage_cache.n_storage_disk;
    // wait for start done
    my_mutex_lock(&g_cache_processor.mutex);
    while(g_cache_processor.fast_disk_open_num != n_fast_disk ||
        g_cache_processor.storage_disk_open_num != n_storage_disk) {
        my_cond_wait(&g_cache_processor.cond, &g_cache_processor.mutex);
    }
    my_mutex_unlock(&g_cache_processor.mutex);

    my_cache_processor_check_and_report();

    g_cache_processor.stats = CACHE_PROCESSOR_STATED;
}

void
my_cache_processor_notify_close(my_bool_t only_close_write) {
    if (only_close_write) {
        g_cache_processor.stats = CACHE_PROCESSOR_CLOSE_WRITE;
    } else {
        g_cache_processor.stats = CACHE_PROCESSOR_CLOSE;
    }

    int n_fast_disk = g_cache_conf.storage_cache.n_fast_disk;
    int n_storage_disk = g_cache_conf.storage_cache.n_storage_disk;

    my_disk_cache_node_t *node;

    for (int i = 0; i < n_fast_disk; i++) {
        node = g_cache_processor.fast_disk_cache_nodes + i;
        if (!node->open) {
            continue;
        }

        my_disk_cache_notify_close(&node->disk_cache, only_close_write);
    }
}


// TODO: 考虑到时间因素, 当前暂不实现 singleflight 功能.
// 因storage层使用了排它锁进行数据保护, 对同一个key并发的读取时存在一些性能瓶颈
// 而singleflight通过复用结果的方式可以提高并发读操作的性能, 并且保证多级缓存各层级将的缓存一致性.

void
my_cache_processor_put(my_cache_req_t *req) {
    if (g_cache_processor.stats != CACHE_PROCESSOR_STATED) {
        req->err = CACHE_ERR_CLOSED;
        req->errmsg = "cache processor write closed or no inited";
        return req->handler(req);
    }

    my_disk_volume_t *volume
        = my_cache_storage_layer_get_volume(g_cache_processor.storage_cache, req->key->key);
    if (!volume) {
        const char *errmsg = "storage cache all offline";
        MY_LOG_ERROR0("cache_processor", errmsg);
        req->err = CACHE_ALL_OFFLINE;
        req->errmsg = errmsg;
        return req->handler(req);
    }

    // call back after waiting for volume processing
    my_disk_volume_put(volume, req);
}

typedef struct {
    my_cache_req_t              *req;
    int                         level;

    // here we want to reuse req, so we need save this context.
    // we will restore it afer use.
    void                        *save_data;         // save req->data
    my_cache_req_handler_pt     save_handler;       // save req->handler
} my_cache_processor_get_ctx_t;

static inline void
my_cache_processor_get_context_init(my_cache_processor_get_ctx_t *ctx, my_cache_req_t *req) {
    bzero(ctx, sizeof(my_cache_processor_get_ctx_t));
    ctx->req = req;
}

static void
my_cache_processor_moveup_done(my_cache_req_t *req);

static inline my_cache_req_t *
my_cache_processor_alloc_new_cache_req(my_cache_req_t *req) {
    my_cache_req_t *new_req = my_malloc(sizeof(my_cache_req_t));
    my_cache_key_t *new_key = my_malloc(sizeof(my_cache_key_t) + req->key->key_len + 1);
    char *key = (char *)(new_key + sizeof(my_cache_key_t));
    memcpy(key, req->key->key, req->key->key_len);
    *(key + req->key->key_len) = '\0';
    my_cache_key_init(new_key, key, req->key->key_len);
    my_buf_chain_t *new_data = my_buf_chain_alloc();
    my_buf_chain_copy_all(new_data, req->data);
    my_cache_req_init(new_req, new_key, new_data, NULL, my_cache_processor_moveup_done);
    return new_req;
}

static inline void
my_cache_processor_free_new_cache_req(my_cache_req_t *new_req) {
    my_free(new_req->key);
    my_buf_chain_free(new_req->data);
    my_free(new_req->data);
    my_free(new_req);
}

static void
my_cache_processor_moveup_done(my_cache_req_t *req) {
    my_cache_processor_free_new_cache_req(req);
}

static void
my_cache_processor_moveup(my_cache_processor_get_ctx_t *ctx) {
    my_cache_req_t *req = ctx->req;

    // no need move up
    if (ctx->level == my_mem_cache_level) {
        return;
    }

    // create new task for move up, need deep copy
    my_cache_req_t *new_req = my_cache_processor_alloc_new_cache_req(req);

    // current level is fast cache, move to mem cache
    if (ctx->level == my_fast_cache_level) {
        my_cache_mem_layer_put(g_cache_processor.mem_cache, new_req->key, new_req->data);
        return my_cache_processor_moveup_done(new_req);
    }

    // current level is storage cache, but no fast cache, move to mem cache
    if (!g_cache_processor.fast_cache) {
        my_cache_mem_layer_put(g_cache_processor.mem_cache, new_req->key, new_req->data);
        return my_cache_processor_moveup_done(new_req);
    }

    my_disk_volume_t *volume
        = my_cache_storage_layer_get_volume(g_cache_processor.fast_cache, req->key->key);
    if (!volume) {
        MY_LOG_ERROR0("cache_processor", "fast cache all offline");
        return my_cache_processor_moveup_done(new_req);
    }

    return my_disk_volume_put(volume, new_req);
}

static void
my_cache_processor_get_done(my_cache_req_t *req) {
    my_cache_processor_get_ctx_t *ctx = req->ctx_data;
    my_cache_processor_moveup(ctx);

    // restorage context
    req->ctx_data = ctx->save_data;
    req->handler = ctx->save_handler;

    my_free(ctx);
    return req->handler(req);
}

static inline void
my_cache_processor_get_in_storage_done(my_cache_req_t *req) {
    my_cache_processor_get_ctx_t *ctx = req->ctx_data;
    if (req->err == CACHE_ERR_NO_ERR) {
        return my_cache_processor_get_done(req);
    }

    // failed to find from storage cache
    if (ctx->level == my_storage_cache_level) {
        return my_cache_processor_get_done(req);
    }

    // search from fast cache, now from storage cache
    ctx->level = my_storage_cache_level;
    my_disk_volume_t *volume =
        my_cache_storage_layer_get_volume(g_cache_processor.storage_cache, req->key->key);
    if (!volume) {
        const char *errmsg = "storage cache all offline";
        req->err = CACHE_ALL_OFFLINE;
        req->errmsg = errmsg;
        return my_cache_processor_get_done(req);
    }

    return my_disk_volume_get(volume, req);
}

void
my_cache_processor_get(my_cache_req_t *req) {
    if (g_cache_processor.stats != CACHE_PROCESSOR_STATED &&
        g_cache_processor.stats != CACHE_PROCESSOR_CLOSE_WRITE) {
        req->err = CACHE_ERR_CLOSED;
        req->errmsg = "cache processor closed or no inited";
        return my_cache_processor_get_done(req);    // failed
    }

    my_cache_processor_get_ctx_t *ctx = my_malloc(sizeof(my_cache_processor_get_ctx_t));
    my_cache_processor_get_context_init(ctx, req);
    // save and switch context
    ctx->save_data = req->ctx_data;
    ctx->save_handler = req->handler;
    req->ctx_data = ctx;
    req->handler = my_cache_processor_get_in_storage_done;

    if (g_cache_processor.mem_cache) {
        ctx->level = my_mem_cache_level;
        if (my_cache_mem_layer_get(g_cache_processor.mem_cache, req->key, req->data)) {
            req->err = CACHE_ERR_NO_ERR;    // success
            return my_cache_processor_get_done(req);
        }
    }

    my_disk_volume_t *volume;
    if (g_cache_processor.fast_cache) {
        ctx->level = my_fast_cache_level;
        volume = my_cache_storage_layer_get_volume(g_cache_processor.fast_cache, req->key->key);
        if (volume) {
            return my_disk_volume_get(volume, req);
        }
        MY_LOG_ERROR0("cache_processor", "fast_cache all offline");
    }

    ctx->level = my_storage_cache_level;
    volume = my_cache_storage_layer_get_volume(g_cache_processor.storage_cache, req->key->key);
    if (!volume) {
        const char *errmsg = "storage cache all offline";
        MY_LOG_ERROR0("cache_processor", errmsg);
        req->err = CACHE_ALL_OFFLINE;
        req->errmsg = errmsg;
        return my_cache_processor_get_done(req);
    }

    return my_disk_volume_get(volume, req);
}
